package belf.migrate.engine.job.structure.helper;

import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.engine.BelfEvn;
import belf.migrate.engine.job.CatalogCreator;
import belf.migrate.api.job.Job;
import belf.migrate.engine.job.JobResult;
import belf.migrate.api.taskconf.Task;
import lombok.extern.slf4j.Slf4j;

/**
 * 处理视图、触发器、存储过程、自定义Class、自定义Type等非表的任务的基类。<br/>
 * 和{@link TableTool}的区别在于jobContext是在构造方法中初始化的。
 */
@Slf4j
public abstract class NonTableTool {
    protected JobContext jobContext;
    protected Job job;
    protected Task task;
    protected TaskConf taskConf;

    /**
     * 单独的任务，仅传入Job对象构造对象，在构造方法力调用{@link #init()}初始化{@Link JobContext}
     * @param job
     */
    public NonTableTool(Job job) {
        this.job = job;
        this.task = job.getTask();
        this.taskConf = job.getTask().getTaskConf();

        init();  //初始化jobContext
    }

    /**
     * 传入{@link Job}和{@link JobContext}参数构造对象，无需再初始化jobContext。一般用于组合任务中，该任务作为子任务被执行
     * @param job
     * @param jobContext
     */
    public NonTableTool(Job job, JobContext jobContext) {
        this.job = job;
        this.jobContext = jobContext;
        this.task = job.getTask();
        this.taskConf = job.getTask().getTaskConf();
    }

    public JobContext getJobContext() {
        return jobContext;
    }

    public Job getJob() {
        return job;
    }

    /**
     * 初始化执行任务需要的环境JobContext
     */
    public void init() {
        jobContext = new JobContext("" + job.getJobId(), taskConf);
        // 创建源端catalog
        AbstractJdbcCatalog SourceCatalog = CatalogCreator.createSourceCatalog(taskConf);
        if (null == SourceCatalog) {
            log.error("No Source Catalog for {}, Stop Job {}!",
                    taskConf.getSourceConf().getCatalogName(),
                    job.getJobId());
        }
        jobContext.setSourceCatalog(SourceCatalog);

        // 创建目标catalog
        AbstractSinkCatalog sinkCatalog = CatalogCreator.createSinkCatalog(taskConf);
        if (null == sinkCatalog) {
            log.error("No Sink Catalog for {}, Stop Job {}!",
                    taskConf.getSinkConf().getCatalogName(),
                    job.getJobId());
        }
        jobContext.setSinkCatalog(sinkCatalog);
        BelfEvn.getInstance().addJobContext("" + job.getJobId(), jobContext);
    }

    public String getSinkQuote() {
        DatabaseType databaseType = DatabaseType.getDatabaseType(jobContext.getTaskConf().getSinkConf().getCatalogName());
        boolean isQuoted = jobContext.getTaskConf().getSinkConf().isQuote();
        return isQuoted ? databaseType.getQuoteType().getValue() : "";
    }

    public QualifiersModel getSourceQualifiersModel() {
        return jobContext.getSourceQualifiersModel();
    }

    public QualifiersModel getSinkQualifiersModel() {
        return jobContext.getSinkQualifiersModel();
    }

    public abstract JobResult run();
}
